package main

import (
	"data-transfer/core"
	"data-transfer/global"
	"data-transfer/plugins"
	m "data-transfer/plugins/mysql"
	"data-transfer/source"
	"data-transfer/transfer"
	"flag"
	_ "github.com/go-sql-driver/mysql"
	"github.com/robfig/cron/v3"
	"go.uber.org/zap"
	"os"
	"time"
)

var configPath string

type transferCmdOption struct {
	TransferType string //操作类型
	BizType      string //需要导入的业务类型：A公告 N资讯 F财务 I资料 S股票涨跌数据
	Biz          string //需要补充数据的具体业务
	BeginDate    string //需要补充数据的起始日期
	EndDate      string //需要补充数据的结束日期
}

var cmdOption transferCmdOption

func init() {

	//1. register mysql plugin
	m.InitPlugin()

	//2. init config
	flag.StringVar(&configPath, "conf", "transfer.toml", "config file")
	flag.StringVar(&cmdOption.TransferType, "transferType", "L", "the transfer type")
	flag.StringVar(&cmdOption.BizType, "bizType", "I", "the load data biz type")
	flag.StringVar(&cmdOption.Biz, "biz", "dstx", "the load data biz")
	flag.StringVar(&cmdOption.BeginDate, "beginDate", "2022-04-21", "the load data begin date")
	flag.StringVar(&cmdOption.EndDate, "endDate", "2022-04-23", "the load data end date")
}

func main() {

	flag.Parse()

	global.GViper = core.Viper(configPath) // 初始化全局用Viper

	global.LOG = core.Zap()                                     //初始化zap log
	global.GBinlogSetPosition = core.GetBinlogSetPosition()     //获取设置的binlog position信息
	global.GBinlogCachePosition = core.GetBinlogCachePosition() //获取缓存的binlog position信息
	//global.ColumnLevel = core.ColumnLevel() //初始化Article的栏目级别

	//1.创建Mysql数据源
	var mysqlDataSource = new(source.MysqlSource)
	dbConfig := mysqlDataSource.Config()

	//2. 取得目标库数据源
	targetStore, err := plugins.NewSQLDB(plugins.DbKindTarget, dbConfig) //F10本地数据源
	if err != nil {
		global.LOG.Fatal("创建F10本地数据源失败！", zap.Any("err", err))
		os.Exit(1)
	} else {
		global.LOG.Info("F10本地数据源创建成功！")
	}
	sourceStore, err := plugins.NewSQLDB(plugins.DbKindSource, dbConfig) //大智慧数据源
	if err != nil {
		global.LOG.Fatal("创建大智慧数据源失败！", zap.Any("err", err))
		os.Exit(1)
	} else {
		global.LOG.Info("大智慧数据源创建成功！")
	}

	canalStore, err := plugins.NewSQLDB(plugins.DbKindCanal, dbConfig) //大智慧数据源,用于goCanal监控
	if err != nil {
		global.LOG.Fatal("创建goCanal监控数据源失败！", zap.Any("err", err))
		os.Exit(1)
	} else {
		global.LOG.Info("goCanal监控数据源创建成功！")
	}

	/*
		infoServiceStore, err := plugins.NewSQLDB(plugins.DbKindInfoService, dbConfig) //infoService 数据源
		if err != nil {
			global.LOG.Fatal("创建infoService数据源失败!", zap.Any("err", err))
		} else {
			global.LOG.Info("infoService数据源创建成功！")
		}
	*/

	// 3.根据命令行参数执行业务
	switch cmdOption.TransferType {
	case transfer.Oper_Init_Test: // 仅仅为测试
		e := transfer.InitTestData(sourceStore, targetStore)
		if e != nil {
			global.LOG.Fatal("初始化测试数据失败！", zap.Any("err", err))
		}
	case transfer.Oper_Init_Announcement: //初始化公告数据
		e := transfer.InitAnnouncementData(sourceStore, targetStore)
		if e != nil {
			global.LOG.Fatal("初始化公告数据失败！", zap.Any("err", err))
		}
	case transfer.Oper_Init_Finance: //初始化财务数据
		e := transfer.InitFinancialData(sourceStore, targetStore)
		if e != nil {
			global.LOG.Fatal("初始化财务数据失败！", zap.Any("err", err))
		}
	case transfer.Oper_Init_SKDAILYPRICE: //初始化股票历史日交易
		e := transfer.InitSkDailyPrice(sourceStore, targetStore)
		if e != nil {
			global.LOG.Fatal("初始化初始化股票历史日交易数据失败！", zap.Any("err", err))
		}
	case transfer.Oper_Init_News: //快讯
		e := transfer.InitNews(sourceStore, targetStore)
		if e != nil {
			global.LOG.Fatal("初始化初始化快讯数据失败！", zap.Any("err", err))
		}
	case transfer.Oper_Init_Info:
		e := transfer.InitInfo(sourceStore, targetStore)
		if e != nil {
			global.LOG.Fatal("初始化初始化资料数据失败！", zap.Any("err", err))
		}
	case transfer.Oper_Monitor: //监听数据库变动
		go AnnounceMtClearCron(targetStore)
		go SyncBinlogPosition(canalStore)
		//handler := &transfer.DataChangeEventHandler{CDB: canalStore, TDB: targetStore, ADB: infoServiceStore, IsFirstRun: false}
		handler := &transfer.DataChangeEventHandler{CDB: canalStore, TDB: targetStore, IsFirstRun: false}
		e := transfer.MonitorDataChange(handler, canalStore, transfer.GoCanal_Run_Flag) //handler 和 canalStore 解耦
		if e != nil {
			global.LOG.Fatal("goCanal监听数据库变动发生错误！", zap.Any("err", err))
		}
	case transfer.Oper_Monitor_FIRST: //监听数据库变动,goCanal第一次执行
		//handler := &transfer.DataChangeEventHandler{CDB: canalStore, TDB: targetStore, ADB: infoServiceStore, IsFirstRun: true}
		handler := &transfer.DataChangeEventHandler{CDB: canalStore, TDB: targetStore, IsFirstRun: true}
		e := transfer.MonitorDataChange(handler, canalStore, transfer.GoCanal_First_Run_Flag) //handler 和 canalStore 解耦
		if e != nil {
			global.LOG.Fatal("goCanal监听数据库变动发生错误！", zap.Any("err", err))
		}

	case transfer.Oper_Init_Load: //补充时间间隔的数据
		if cmdOption.BizType == "" {
			global.LOG.Fatal("请输入需要补充数据的业务类型：-bizType A公告 N资讯 F财务 I资料 S股票涨跌数据")
		}
		if cmdOption.Biz == "" {
			global.LOG.Fatal("请输入需要补充数据的具体业务！")
		}

		if cmdOption.BeginDate == "" {
			global.LOG.Fatal("请输入需要补充数据的业务起始日期，格式：2022-02-07")
		}
		if cmdOption.EndDate == "" {
			global.LOG.Fatal("请输入需要补充数据的业务结束日期，格式：2022-02-08")
		}

		_, errf := time.Parse("2006-01-02", cmdOption.BeginDate)
		if errf != nil {
			global.LOG.Fatal("请输入正确格式的起始日期，格式：2022-02-07")
		}

		_, errf = time.Parse("2006-01-02", cmdOption.EndDate)
		if errf != nil {
			global.LOG.Fatal("请输入正确格式的结束日期，格式：2022-02-07")
		}

		beginDate := cmdOption.BeginDate + " 00:00:00"
		endDate := cmdOption.EndDate + " 00:00:00"

		transfer.LoadData(sourceStore, targetStore, cmdOption.BizType, cmdOption.Biz, beginDate, endDate)

	default:
		global.LOG.Info("......没有进行任何操作......")
	}
}

// SyncBinlogPosition 定时同步Mysql binlog 的position
func SyncBinlogPosition(m plugins.DataMonitor) {
	c := cron.New()
	//每分钟执行一次
	_, err := c.AddFunc("*/1 * * * *", func() {
		m.SyncBinlogPosition()
	})

	if err != nil {
		global.LOG.Error("公告数据清理定时任务构造失败:", zap.Any("err", err))
	}

	c.Start()
}

//AnnounceMtClearCron 公告数据清理定时任务，每天执行一次，清理一年前的数据
func AnnounceMtClearCron(t plugins.TargetStore) {
	c := newWithSeconds()

	_, err := c.AddFunc("0 0 23 * * ?", func() {
		t.ClearAnnouncement()
	})

	if err != nil {
		global.LOG.Error("公告数据清理定时任务构造失败:", zap.Any("err", err))
	}

	c.Start()

}

// 返回一个支持至 秒 级别的 cron
func newWithSeconds() *cron.Cron {
	secondParser := cron.NewParser(cron.Second | cron.Minute |
		cron.Hour | cron.Dom | cron.Month | cron.DowOptional | cron.Descriptor)
	return cron.New(cron.WithParser(secondParser), cron.WithChain())
}
